﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading;

namespace ZHello.Async
{
    public static class ProducerConsumer
    {
        /// <summary>
        /// 一个生产者一个消费者
        /// </summary>
        public static void Main()
        {
            var random = new Random(17);
            long lid = 0;
            var test = new ProducerConsumer<string>(5)
            {
                ProductFunc = new Func<string>(() =>
                {
                    lid++;
                    var ret = string.Format("【{0}】", lid);
                    var info = string.Format("{0} 生产：{1}", Thread.CurrentThread.ManagedThreadId, ret);
                    Trace.WriteLine(info);
                    Console.WriteLine(info);
                    return ret;
                }),
                ConsumeFunc = new Action<string>((str) =>
                {
                    var ret = string.Format("{0} 消费:{1}", Thread.CurrentThread.ManagedThreadId, str);
                    Console.WriteLine(ret);
                    Trace.WriteLine(ret);
                })
            };

            var th1 = new Thread(() =>
            {
                while (true)
                {
                    Thread.Sleep(random.Next(100, 300));
                    test.Product();
                }
            });
            var th2 = new Thread(() =>
            {
                while (true)
                {
                    Thread.Sleep(random.Next(50, 60));
                    test.Consume();
                }
            });

            test.Product();
            test.Product();
            test.Product();
            test.Product();
            test.Product();

            th1.Start();
            th2.Start();

            Thread.Sleep(60000);

            th1.Abort();
            th2.Abort();
        }

        /// <summary>
        /// 一个生产者多个消费者
        /// </summary>
        public static void Main1_N()
        {
            var random = new Random(17);
            long lid = 0;
            var test = new ProducerConsumer<string>(5)
            {
                ProductFunc = new Func<string>(() =>
                {
                    lid++;
                    var ret = string.Format("【{0}】", lid);
                    var info = string.Format("{0} 生产：{1}", Thread.CurrentThread.ManagedThreadId, ret);
                    Trace.WriteLine(info);
                    Console.WriteLine(info);
                    return ret;
                }),
                ConsumeFunc = new Action<string>((str) =>
                {
                    var ret = string.Format("{0} 消费:{1}", Thread.CurrentThread.ManagedThreadId, str);
                    Console.WriteLine(ret);
                    Trace.WriteLine(ret);
                })
            };
            var th1 = new Thread(() =>
            {
                while (true)
                {
                    Thread.Sleep(random.Next(50, 60));
                    test.Product();
                }
            });
            var ths = new List<Thread>();
            for (int i = 0; i < 4; i++)
            {
                var th = new Thread(() =>
                {
                    while (true)
                    {
                        Thread.Sleep(random.Next(500, 600));
                        test.Consume();
                    }
                });
                ths.Add(th);
            }

            test.Product();
            test.Product();
            test.Product();
            test.Product();
            test.Product();
            th1.Start();

            ths.ForEach(t => t.Start());

            Thread.Sleep(60000 * 10);

            th1.Abort();
            ths.ForEach(t => t.Abort());
        }

        /// <summary>
        /// 多个生产者一个消费者
        /// </summary>
        public static void MainN_1()
        {
            var random = new Random(17);
            long lid = 0;
            var test = new ProducerConsumer<string>(5)
            {
                ProductFunc = new Func<string>(() =>
                {
                    lid++;
                    var ret = string.Format("【{0}】", lid);
                    var info = string.Format("{0} 生产：{1}", Thread.CurrentThread.ManagedThreadId, ret);
                    Trace.WriteLine(info);
                    Console.WriteLine(info);
                    return ret;
                }),
                ConsumeFunc = new Action<string>((str) =>
                {
                    var ret = string.Format("{0} 消费:{1}", Thread.CurrentThread.ManagedThreadId, str);
                    Console.WriteLine(ret);
                    Trace.WriteLine(ret);
                })
            };
            var thps = new List<Thread>();
            for (int i = 0; i < 3; i++)
            {
                var th = new Thread(() =>
                {
                    while (true)
                    {
                        Thread.Sleep(random.Next(10, 30));
                        test.ProductN();
                    }
                });
                thps.Add(th);
            }
            var th2 = new Thread(() =>
            {
                while (true)
                {
                    Thread.Sleep(random.Next(10, 30));
                    test.ConsumeN();
                }
            });

            th2.Start();
            thps.ForEach(t => t.Start());

            Thread.Sleep(60000 * 10);

            th2.Abort();
            thps.ForEach(t => t.Abort());
        }

        /// <summary>
        /// 多个生产者多个消费者
        /// </summary>
        public static void MainN_N()
        {
            var random = new Random(17);
            long lid = 0;
            var test = new ProducerConsumer<string>(5)
            {
                ProductFunc = new Func<string>(() =>
                {
                    lid++;
                    var ret = string.Format("【{0}】", lid);
                    var info = string.Format("{0} 生产:{1}", Thread.CurrentThread.ManagedThreadId, ret);
                    Trace.WriteLine(info);
                    Console.WriteLine(info);
                    return ret;
                }),
                ConsumeFunc = new Action<string>((str) =>
                {
                    var ret = string.Format("{0} 消费:{1}", Thread.CurrentThread.ManagedThreadId, str);
                    Console.WriteLine(ret);
                    Trace.WriteLine(ret);
                })
            };
            var thps = new List<Thread>();
            for (int i = 0; i < 3; i++)
            {
                var th = new Thread(() =>
                {
                    while (true)
                    {
                        Thread.Sleep(random.Next(500, 600));
                        test.ProductN();
                    }
                });
                thps.Add(th);
            }
            var ths = new List<Thread>();
            for (int i = 0; i < 4; i++)
            {
                var th = new Thread(() =>
                {
                    while (true)
                    {
                        Thread.Sleep(random.Next(50, 60));
                        test.ConsumeN();
                    }
                });
                ths.Add(th);
            }

            thps.ForEach(t => t.Start());
            ths.ForEach(t => t.Start());

            Thread.Sleep(60000 * 10);

            thps.ForEach(t => t.Abort());
            ths.ForEach(t => t.Abort());
        }
    }

    public class ProducerConsumer<TProduct> : IDisposable
    {
        public ProducerConsumer(int capacity)
        {
            if (capacity <= 0)
                throw new ArgumentException();
            Capacity = capacity;
            Products = new Queue<TProduct>(capacity);
            WriteSemaphore = new Semaphore(capacity, capacity);
            ReadSemaphore = new Semaphore(0, capacity);
        }

        /// <summary>
        /// 库存队列
        /// </summary>
        private Queue<TProduct> Products;

        /// <summary>
        /// 库存容量
        /// </summary>
        private int Capacity;

        /// <summary>
        /// 生产产品的方法
        /// </summary>
        public Func<TProduct> ProductFunc;

        /// <summary>
        /// 消费产品的方法
        /// </summary>
        public Action<TProduct> ConsumeFunc;

        private bool IsDesposed = false;
        private object _disposeObj = new object();

        public void Dispose()
        {
            if (IsDesposed)
                return;
            lock (_disposeObj)
            {
                if (IsDesposed)
                    return;
                IsDesposed = true;
                WriteEvent.Close();
                ReadEvent.Close();
                WriteSemaphore.Close();
                ReadSemaphore.Close();
            }
        }

        #region 单生产者-单消费者

        /// <summary>
        /// 生产者入队锁定对象，消费者判空等待对象
        /// </summary>
        private object objP0 = new object();

        /// <summary>
        /// 消费者出对锁定对象，生产者判满等待对象
        /// </summary>
        private object objC0 = new object();

        /// <summary>
        /// 三个阶段
        /// 1.库存判断（库存已满的情况下需等待）
        /// 2.生产产品
        /// 3.产品入库
        /// </summary>
        public void Product()
        {
            if (IsDesposed)
                return;
            //产品已满判断
            Monitor.Enter(objC0);
            var count = Products.Count;
            if (count > Capacity)
                Console.WriteLine("***Exception OverFull***");
            if (count == Capacity)
            {
                var str = string.Format("队列已满({0})，{1}等待消费", count, Thread.CurrentThread.ManagedThreadId);
                Trace.Write(str);
                Console.WriteLine(str);
                //等待产品消费
                Monitor.Wait(objC0);
            }
            Monitor.Exit(objC0);
            //生产产品
            var product = ProductFunc.Invoke();
            //开始入队产品，获取独占资源
            Monitor.Enter(objP0);
            //入队期间，该生产者独占产品队列
            Products.Enqueue(product);
            //入队结束，通知等待产品的消费者，产品队列增加
            Monitor.Pulse(objP0);
            //入队结束，释放独占队列
            Monitor.Exit(objP0);
        }

        /// <summary>
        /// 三个阶段
        /// 1.库存判断（产品无货情况下需等待）
        /// 2.产品出库
        /// 3.产品消费
        /// </summary>
        public void Consume()
        {
            if (IsDesposed)
                return;
            //产品无货判断
            Monitor.Enter(objP0);
            var count = Products.Count;
            if (count == 0)
            {
                var str = string.Format("队列已空({0})，{1}等待生产", count, Thread.CurrentThread.ManagedThreadId);
                Trace.Write(str);
                Console.WriteLine(str);
                //等待产品生产
                Monitor.Wait(objP0);
            }
            Monitor.Exit(objP0);
            //开始取出产品，获取独占资源
            Monitor.Enter(objC0);
            //取出期间，该消费者独占产品队列，即此时不能判断产品缓存是否已满
            var product = Products.Dequeue();
            //取出完毕，通知等待生产的生产者，产品队列减少
            Monitor.Pulse(objC0);
            //取出完毕，释放独占资源
            Monitor.Exit(objC0);
            //产品消费
            ConsumeFunc.Invoke(product);
        }

        #endregion 单生产者-单消费者

        #region 多生产者-多消费者

        /// <summary>
        /// 生产时，满库时等待信号
        /// </summary>
        private ManualResetEvent WriteEvent = new ManualResetEvent(false);

        /// <summary>
        /// 消费时，空库时等待信号
        /// </summary>
        private ManualResetEvent ReadEvent = new ManualResetEvent(false);

        /// <summary>
        /// 可写的数量
        /// </summary>
        private Semaphore WriteSemaphore;

        /// <summary>
        /// 可读的数量
        /// </summary>
        private Semaphore ReadSemaphore;

        /// <summary>
        /// 三个阶段
        /// 1.库存判断（库存已满的情况下需等待）多个线程可以同时等待以及查询
        /// 2.生产产品
        /// 3.产品入库
        /// </summary>
        public void ProductN()
        {
            if (IsDesposed)
                return;
            //产品已满判断
            int count = Products.Count;
            if (count == Capacity)
            {
                WriteEvent.Reset();
                var str = string.Format("队列已满({0})，{1}等待消费", count, Thread.CurrentThread.ManagedThreadId);
                Trace.Write(str);
                Console.WriteLine(str);
                if (count > Capacity)
                    Console.WriteLine("***Exception OverFull***");
                //等待产品消费
                WriteEvent.WaitOne();
            }
            //消耗一个写入资源
            if (!WriteSemaphore.WaitOne())
            {
                ProductN();
                return;
            }
            //生产产品
            var product = ProductFunc.Invoke();
            //开始入队产品
            Products.Enqueue(product);
            //入队结束，通知等待产品的消费者，产品队列增加，释放一个可用的读取信号资源
            ReadSemaphore.Release();
            ReadEvent.Set();
        }

        public void ConsumeN()
        {
            if (IsDesposed)
                return;
            //产品无货判断
            int count = Products.Count;
            if (count == 0)
            {
                ReadEvent.Reset();
                var str = string.Format("队列已空({0})，{1}等待生产", count, Thread.CurrentThread.ManagedThreadId);
                Trace.Write(str);
                Console.WriteLine(str);
                //等待产品生产
                ReadEvent.WaitOne();
            }
            //消耗一个读取信号资源
            if (!ReadSemaphore.WaitOne())
            {
                ConsumeN();
                return;
            }
            //开始取出产品
            var product = Products.Dequeue();
            //取出完毕，通知等待生产的生产者，产品队列减少，释放一个可用的写入信号资源
            WriteSemaphore.Release();
            WriteEvent.Set();
            //产品消费
            ConsumeFunc.Invoke(product);
        }

        #endregion 多生产者-多消费者
    }
}